/*  g++ test_lock.cc -o test_lock -std=c++11 -lpthread
 * */

#include <thread>
#include <mutex>
#include <vector>
#include <iostream>
#include <unistd.h>
#include <algorithm>
#include <atomic>
#include <assert.h>
#include <functional>
#include <chrono>
#include <condition_variable>    // std::condition_variable

//std::mutex my_lock_empty;
//std::mutex my_lock_full;

std::mutex my_lock;
std::condition_variable cv_empty;
std::condition_variable cv_full;

std::atomic_int count{0};

class ShardQueue{
   public:
    ShardQueue(){}
    const int q_num = 4
    std::vector<int> q[q_num];
    std::mutex qlock[q_num];
    int MAX_LEN=1024;

}
double runtime = 10000000 ;

auto start = std::chrono::steady_clock::now();

double get_duration()
{
	auto end = std::chrono::steady_clock::now();
	std::chrono::duration<double, std::micro> elapsed = end - start; // std::micro 表示以微秒为时间单位
	//std::cout<< "time: "  << elapsed.count() << "us" << std::endl;
	return elapsed.count();
}


void do_process()
{
      //usleep(1);		
      int sum;
      for(int i=0; i<100; i++)
	      for(int j=0; j<100; j++)
		      sum = i * 10 + j;
}
void consumer(int& i){
	std::vector<int> local_q;
	while( true ){
		std::unique_lock<std::mutex> lock(my_lock); 
		if(q.empty()){
		   if( get_duration() >= runtime)
			   break;
                   cv_empty.wait(lock);
		}
		else{
			local_q.swap(q);
			lock.unlock();
			cv_full.notify_all();

			std::cout<<"size:" << local_q.size() << std::endl;
			while(local_q.size()){
				int item = local_q.front();
				do_process();
				local_q.erase(local_q.begin());
				count++;
				//if(count % 100000 == 0)
				//	std::cout<<"consumer item: "<< count << std::endl;
			}
		}
	}   
}

void product(int& i)
{
	while(true){
		std::unique_lock<std::mutex> lock(my_lock);
		if(q.size() == MAX_LEN){
			if( get_duration() >= runtime)
				break;
			cv_full.wait(lock);
		}else{
			std::clock_t end = std::clock();
			//std::cout<< "sleep: "<< end - start << " ms"<<std::endl;
			int item = rand();
			//	sleep_for(chrono::milliseconds(10));  
			q.push_back(item);
			//std::cout<<"product item: " << item << std::endl;
			lock.unlock();  
			cv_empty.notify_all(); //cv.notify_all();
		}
	}
}


int main(){
	int num = 0;
	int num_thread=8;
	std::vector<std::thread> ver;   //保存线程的vector
	for(int i = 0; i < num_thread; ++i){
		std::thread t = std::thread(product,std::ref(num));
		std::string name= "product" + std::to_string(i);
		//set_thread_name(t,name);
		int ret = pthread_setname_np(t.native_handle(), name.c_str());
		assert(ret==0);
		ver.emplace_back(std::move(t)); //保存线程
	}   
	for(int i = 0; i < num_thread; ++i){
		std::thread t = std::thread(consumer, std::ref(num));
		std::string name= "consumer" + std::to_string(i);
		int ret=pthread_setname_np(t.native_handle(), name.c_str());
		assert(ret==0);
		ver.emplace_back(std::move(t)); //保存线程
	}

	std::for_each(ver.begin(), ver.end(), std::mem_fn(&std::thread::join)); //join
	std::cout << "count: "  << count <<  std::endl;
}
